package com.navi.job.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.navi.entity.ErrorMessageInfoEntity;
import com.navi.entity.HisConfigure;
import com.navi.entity.TableEntity;
import com.navi.exception.HmsException;
import com.navi.service.IGploaderService;
import com.navi.service.impl.GploaderService;
import com.navi.service.impl.TableMetaService;
import com.navi.util.*;
import com.rabbitmq.client.*;
import com.sun.org.apache.bcel.internal.generic.IF_ACMPEQ;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.IOException;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

import static com.navi.constant.GenericDef.*;
import static com.navi.constant.GenericDef.LOG_SUB_MAIN_PROC_SUM;
import static java.util.stream.Collectors.toList;

@Service
@Slf4j
public class MainService {

    @Value("${spring.rabbitmq.exchange}")
    private String exchange;
    @Value("${spring.rabbitmq.error.queue.routekey}")
    private String errorQueueRouteKey;
    @Value("${spring.rabbitmq.error.queue.routekey2}")
    private String errorQueueRouteKey2;


    @Value("${app.conf.table.eda_pds_glass_arycf}")
    private String edaPdsGlassArycf;
    @Value("${app.conf.table.eda_pds_glass_cell1}")
    private String edaPdsGlassCell1;
    @Value("${app.conf.table.eda_pds_glass_cell2}")
    private String edaPdsGlassCell2;
    @Value("${app.conf.table.eda_pds_glass_summary_arycf}")
    private String edaPdsGlassSummaryArycf;
    @Value("${app.conf.table.eda_pds_glass_summary_cell1}")
    private String edaPdsGlassSummaryCell1;
    @Value("${app.conf.table.eda_pds_glass_summary_cell2}")
    private String edaPdsGlassSummaryCell2;
    @Value("${app.conf.table.eda_pds_result_arycf}")
    private String edaPdsResultArycf;
    @Value("${app.conf.table.eda_pds_result_cell1}")
    private String edaPdsResultCell1;
    @Value("${app.conf.table.eda_pds_result_cell2}")
    private String edaPdsResultCell2;

    @Value("${app.conf.table.eda_mds_glass_arycf}")
    private String edaMdsGlassArycf;
    @Value("${app.conf.table.eda_mds_glass_cell1}")
    private String edaMdsGlassCell1;
    @Value("${app.conf.table.eda_mds_glass_cell2}")
    private String edaMdsGlassCell2;
    @Value("${app.conf.table.eda_mds_glass_summary_arycf}")
    private String edaMdsGlassSummaryArycf;
    @Value("${app.conf.table.eda_mds_glass_summary_cell1}")
    private String edaMdsGlassSummaryCell1;
    @Value("${app.conf.table.eda_mds_glass_summary_cell2}")
    private String edaMdsGlassSummaryCell2;
    @Value("${app.conf.table.eda_mds_result_arycf}")
    private String edaMdsResultArycf;
    @Value("${app.conf.table.eda_mds_result_cell1}")
    private String edaMdsResultCell1;
    @Value("${app.conf.table.eda_mds_result_cell2}")
    private String edaMdsResultCell2;

    @Autowired
    private JedisPool jedisPool;

    @Autowired
    private TestConnectionUtil testConnectionUtil;

    @Autowired
    private IGploaderService gploaderService;

    @Autowired
    private ConnectionFactory connectionFactory;


    /**
     * 表主键的 字段名vs驼峰命名
     */
    private static Map<String, Map<String, String>> primaryColumnCamelMap = new HashMap();


//    public Integer subMainProc(String queueName, String tableName, List<JSONObject> jsonObjectList, Channel channel, Map<GetResponse, String> responseMap, String limit_time, String deleteInsert, String jobBatchId) throws Exception {
//        List<JSONObject> updateList = new ArrayList<>();
//        List<JSONObject> realFlgList = new ArrayList<>();
//        try (Jedis jedis = jedisPool.getResource()) {
//            //region  get redis lock
//
//            String tableLockKey = String.format(REDIS_LOCK_KEY_FORMAT, tableName);
//            // 1 if the key was set 0 if the key was not set
//            Long getRedisLock = jedis.setnx(tableLockKey, queueName);
//            String curLockVal = jedis.get(tableLockKey);
//            if (getRedisLock == 1) {
//                log.info("{} queue:{}, table:{} 获取锁[{}]成功, 当前锁值[{}].",
//                        REDIS_OPE, queueName, tableName, tableLockKey, curLockVal);
//            } else {
//                if (queueName.equals(curLockVal)) {
//                    log.info("{} 是当前queue {} 的锁，继续执行, 当前锁值[{}].", REDIS_OPE, queueName, curLockVal);
//                } else {
//                    log.info("{} queue:{}, table:{} 获取锁[{}]失败, 当前锁值[{}].",
//                            REDIS_OPE, queueName, tableName, tableLockKey, curLockVal);
//                    return RETURN_REWORK_THISCASE;
//                }
//            }
//            //endregion
//        } catch (Exception e) {
//            log.error("HMS送往Redis key失败，原因{}",LogUtil.stackTraceToString(e));
//            return RETURN_REWORK_THISCASE;
//        }
//
//        // 更新的Q抽出来，目前兼容新老两种，便于切换
//        final long allRowCount = jsonObjectList.size();
//        updateList = jsonObjectList.stream().filter(jsonObject -> jsonObject.containsKey("update_flg")).collect(toList());
//        jsonObjectList.removeAll(updateList);
//        realFlgList = jsonObjectList.stream().filter(jsonObject -> jsonObject.containsKey("real_flg")).collect(toList());
//        jsonObjectList.removeAll(realFlgList);
//
//        // region Process one table message
//        try (final java.sql.Connection dbConnection = testConnectionUtil.getDefaultConnection()) {
//            long beginTime2 = System.currentTimeMillis();
//            long tableProcStartTime = beginTime2;
//            boolean stageAlreadyCreateFlg = false;
//
//            log.info("records from queue {} ", jsonObjectList.size());
//
//            //从redis从得到配置文件
//            //从配置文件中 读取 此文件是要 insert 还是要merge ，以及目标表的名称和schema
//            try (Jedis jedis = jedisPool.getResource()) {
//                String strHisConfig = jedis.hget(REDIS_HIS_CONFIG, tableName);
//
//                //region process config from redis
//                /**
//                 *  处理批量设定
//                 *  多张表对应同一个设定 edc_data_* 。 defect子分区表。 多张表对应同一个 eda_*
//                 *  后续可以在增加一类模糊匹配设定，找不到先去找模糊匹配的设定
//                 */
//                //result
//                if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_CELL2);
//                }
//                //summary
//                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL2);
//                }
//                //glass
//                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_CELL2);
//                }
//
//                //mds
//                //result
//                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_CELL2);
//                }
//                //summary
//                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL2);
//                }
//                //glass
//                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassArycf.split(","))//先summary 后 glass
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_ARYCF);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassCell1.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_CELL1);
//                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassCell2.split(","))
//                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
//                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_CELL2);
//                }
//
//
//                //如果找不到对应的配置文件，那么直接送ERROR_QUEUE
//                if (StringUtils.isEmpty(strHisConfig)) {
//                    log.error("table {} can not find config file", tableName);
//                    sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                    return RETURN_FAISE;
//                }
//                HisConfigure hisConfigure = JacksonUtil.fromJson(strHisConfig, HisConfigure.class);
//                if (null != hisConfigure) {
//                    hisConfigure.setTableName(tableName);
//                    if (hisConfigure.getPptFlg()) {
//                        hisConfigure.setPptTableName(tableName);
//                    }
//                    if (hisConfigure.getHisFlg()) {
//                        hisConfigure.setHisTableName(tableName);
//                    }
//                }
//
//                //如果配置文件不是正确的格式，也直接送ERROR_QUEUE
//                if (hisConfigure == null) {
//                    log.error("table {}  config can not parse correctly", tableName);
//                    sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                    return RETURN_FAISE;
//                }
//                //endregion
//
//                //region merge insert update
//                String stageTableName = String.format("templates11_%s", tableName);
//                String extTableName = String.format("ext_%s", tableName);
//
//                if (!jsonObjectList.isEmpty()) {
//                    if (hisConfigure.getPptFlg()) {
//                        final String pptTableSchema = hisConfigure.getPptTableSchema();
//                        final String pptTableName = hisConfigure.getPptTableName();
//                        final int effectRows = gploaderService.mergeProc(dbConnection, pptTableSchema, pptTableName, stageTableName, extTableName, jsonObjectList, limit_time, deleteInsert);
//                        final String msg = "写入PPT表[" + pptTableName + "],记录数:" + effectRows;
//                        stageAlreadyCreateFlg = true;
//                        LogUtil.printPerformance(LOG_MERGE_PROC, queueName, jsonObjectList.size(), beginTime2);
//                        beginTime2 = LogUtil.printCost(LOG_MERGE_PROC, queueName, msg, beginTime2);
//                        stageTableName = String.format("templates11_%s", tableName);
//                    }
//
//                    log.info("==========================================");
//
//                    if (hisConfigure.getHisFlg()) {
//                        final String hisTableSchema = hisConfigure.getHisTableSchema();
//                        final String hisTableName = hisConfigure.getHisTableName();
//                        final int effectRows = gploaderService.insertProc(dbConnection, hisTableSchema, hisTableName, stageTableName, extTableName, jsonObjectList, stageAlreadyCreateFlg);
//                        final String msg = "写入HIS表[" + hisTableName + "],记录数:" + effectRows;
//                        LogUtil.printPerformance(LOG_SUB_MAIN_PROC_HIS, queueName, jsonObjectList.size(), beginTime2);
//                        beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_HIS, queueName, msg, beginTime2);
//                    }
//                } else {
//                    log.info("jsonObjectList is empty, updateList size is: {}.", updateList.size());
//                }
//
//                //update
//                if (CollectionUtils.isNotEmpty(updateList)) {
//                    final String pptTableSchema = hisConfigure.getPptTableSchema();
//                    final String pptTableName = hisConfigure.getPptTableName();
//                    final int effectRows = gploaderService.updateProc(dbConnection, jedis, updateList, pptTableSchema, pptTableName, stageTableName);
//                    final String msg = "更新HIS表[" + pptTableName + "],记录数:" + effectRows;
//                    LogUtil.printPerformance(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, updateList.size(), beginTime2);
//                    beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, msg, beginTime2);
//                }
//
//                // 更新 real_flg 切换期间兼容
//                if (CollectionUtils.isNotEmpty(realFlgList)) {
//                    log.warn("进入了老的 real_flg 更新逻辑，请尽快修改，table: {}", tableName);
//                    final int effectRows = gploaderService.updateProc(dbConnection, realFlgList);
//                    final String msg = "更新HIS表[" + tableName + "],记录数:" + effectRows;
//                    LogUtil.printPerformance(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, realFlgList.size(), beginTime2);
//                    beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, msg, beginTime2);
//                }
//                //endregion
//
//                //确定消费
//                // 这里其实可以拿到TableSchema， 但是如果同时有PPT/HIS两种配置的话程序需要调整，目前使用默认操作
//                sureAck(channel, queueName, tableName, responseMap);
//                LogUtil.printPerformance(ACK, tableName, responseMap.size(), beginTime2);
//                beginTime2 = LogUtil.printCost(ACK, queueName,
//                        "sureAck:" + tableName + ", responseMap: " + responseMap.size(), beginTime2);
//            } catch (SQLException sqlEx) {
//                log.error("SQL exception: SQL State [{}], Error Code [{}], Error Message [{}].",
//                        sqlEx.getSQLState(), sqlEx.getErrorCode(), sqlEx.getMessage());
//                LogUtil.pringStackTrace(sqlEx);
//                throw sqlEx;
//            } catch (Exception ex3) {
//                LogUtil.pringStackTrace(ex3);
//                throw ex3;
//            }
//
//            log.info("table_name;{} total cost {} millis", tableName, System.currentTimeMillis() - tableProcStartTime);
//            LogUtil.printPerformance(LOG_SUB_MAIN_PROC_SUM, tableName, allRowCount, tableProcStartTime);
//            LogUtil.printCost(LOG_SUB_MAIN_PROC_SUM, tableName, " count:" + allRowCount, tableProcStartTime);
//
//        } catch (SQLException sqlEx) {
//            LogUtil.pringStackTrace(sqlEx);
//            try {
//                if (catchGpConnectionException(sqlEx.toString())) {
//                    log.error("[][][] 出现gp连接异常，Que不消费...");
//                } else {
//                    sendJsonToErrorQueue(errorQueueRouteKey, queueName, tableName, sqlEx, null, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                }
//            } catch (Exception e) {
//                LogUtil.pringStackTrace(e);
//            }
//        } catch (Exception ex) {
//            /**
//             * 如果一张表的数据没有处理完成，只送这一张表的ErrorQueue
//             * 其余表对应的消息应该不受影响
//             * 在此处送ERROR_QUEUE
//             */
//            LogUtil.pringStackTrace(ex);
//            try {
//                if (catchGpConnectionException(ex.toString())) {
//                    log.error("[][][] 出现gp连接异常，Que不消费...");
//                } else {
//                    sendJsonToErrorQueue(errorQueueRouteKey, queueName, tableName, null, ex, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                }
//            } catch (Exception e) {
//                LogUtil.pringStackTrace(e);
//            }
//        }
//        //endregion
//
//        return RETURN_SUCCESS;
//        //结束:打印耗时，写入行数等性能参数
//    }


    public Integer subMainProcForPds(String queueName, String tableName, List<JSONObject> jsonObjectList, Channel channel, Map<GetResponse, String> responseMap, String limit_time, String deleteInsert, String jobBatchId) throws Exception {
        List<JSONObject> updateList = new ArrayList<>();
        List<JSONObject> realFlgList = new ArrayList<>();
        try (Jedis jedis = jedisPool.getResource()) {
            //region  get redis lock

            String tableLockKey = String.format(REDIS_LOCK_KEY_FORMAT, tableName);
            // 1 if the key was set 0 if the key was not set
            Long getRedisLock = jedis.setnx(tableLockKey, queueName);
            String curLockVal = jedis.get(tableLockKey);
            if (getRedisLock == 1) {
                log.info("{} queue:{}, table:{} 获取锁[{}]成功, 当前锁值[{}].",
                        REDIS_OPE, queueName, tableName, tableLockKey, curLockVal);
            } else {
                if(null==curLockVal){
                    jedis.del(tableLockKey);
                    log.info("{} queue:{}, table:{} 获取锁[{}]异常, 当前锁值[{}],为NULL. 删除该锁，重启...",
                            REDIS_OPE, queueName, tableName, tableLockKey, curLockVal);
                    return RETURN_REWORK_THISCASE;
                }
                if (queueName.equals(curLockVal)) {
                    log.info("{} 是当前queue {} 的锁，继续执行, 当前锁值[{}].", REDIS_OPE, queueName, curLockVal);
                } else {
                    log.info("{} queue:{}, table:{} 获取锁[{}]失败, 当前锁值[{}].",
                            REDIS_OPE, queueName, tableName, tableLockKey, curLockVal);
                    if ((tableName.contains("summary") || tableName.contains("result")) && (tableName.contains("pds") || tableName.contains("mds"))) {
                        throw new Exception("表" + tableName + " 获取锁失败，不执行glass部分。");
                    }
                    return RETURN_REWORK_THISCASE;
                }
            }
            //endregion
        } catch (Exception e) {
            log.error("HMS送往Redis key失败，原因{}", LogUtil.stackTraceToString(e));
            return RETURN_REWORK_THISCASE;
        }

        // 更新的Q抽出来，目前兼容新老两种，便于切换
        final long allRowCount = jsonObjectList.size();
        updateList = jsonObjectList.stream().filter(jsonObject -> jsonObject.containsKey("update_flg")).collect(toList());
        jsonObjectList.removeAll(updateList);
        realFlgList = jsonObjectList.stream().filter(jsonObject -> jsonObject.containsKey("real_flg")).collect(toList());
        jsonObjectList.removeAll(realFlgList);

        // region Process one table message
        try (final java.sql.Connection dbConnection = testConnectionUtil.getDefaultConnection()) {
            long beginTime2 = System.currentTimeMillis();
            long tableProcStartTime = beginTime2;
            boolean stageAlreadyCreateFlg = false;

            log.info("records from queue {} ", jsonObjectList.size());

            //从redis从得到配置文件
            //从配置文件中 读取 此文件是要 insert 还是要merge ，以及目标表的名称和schema
            try (Jedis jedis = jedisPool.getResource()) {
                String strHisConfig = jedis.hget(REDIS_HIS_CONFIG, tableName);

                //region process config from redis
                /**
                 *  处理批量设定
                 *  多张表对应同一个设定 edc_data_* 。 defect子分区表。 多张表对应同一个 eda_*
                 *  后续可以在增加一类模糊匹配设定，找不到先去找模糊匹配的设定
                 */
                //result
                if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsResultCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_RESULT_TABLE_DEFAULT_CONF_CELL2);
                }
                //summary
                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassSummaryCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL2);
                }
                //glass
                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaPdsGlassCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_PDS_GLASS_TABLE_DEFAULT_CONF_CELL2);
                }

                //mds
                //result
                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsResultCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_RESULT_TABLE_DEFAULT_CONF_CELL2);
                }
                //summary
                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassSummaryCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_SUMMARY_TABLE_DEFAULT_CONF_CELL2);
                }
                //glass
                else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassArycf.split(","))//先summary 后 glass
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_ARYCF);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassCell1.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_CELL1);
                } else if (StringUtils.isEmpty(strHisConfig) && (Arrays.asList(edaMdsGlassCell2.split(","))
                        .stream().filter(ele -> tableName.startsWith(ele)).count() > 0)) {
                    strHisConfig = jedis.hget(REDIS_HIS_CONFIG, EDA_MDS_GLASS_TABLE_DEFAULT_CONF_CELL2);
                }


                //如果找不到对应的配置文件，那么直接送ERROR_QUEUE
                if (StringUtils.isEmpty(strHisConfig)) {
                    log.error("table {} can not find config file", tableName);
//                    sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);

//                    try {
//                        for(JSONObject updateObj : jsonObjectList) {
//                            if (StringUtils.isNotEmpty(updateObj.getString("query_conditions"))) {
//                                JSONObject jsonObject = (JSONObject) JSONObject.parse(updateObj.getString("query_conditions"));
//                                String opeId = jsonObject.getString("ope_id");
//                                String glassId = jsonObject.getString("glass_id");
//                                String evtTimestampStr = jsonObject.getString("evt_timestamp");
//                                String jgeCnt = jsonObject.getString("jge_cnt");
//                                try {
//                                    JSONObject jsonObjNew = new JSONObject();
//                                    jsonObjNew.put("dataInfo", String.format("opeId:%s glassId:%s evtTimestamp:%s jgeCnt:%s找不到对应的配置文件", opeId, glassId, evtTimestampStr, jgeCnt));
//                                    jsonObjNew.put("dataName", "");
//
//                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
//                                    String unqSeqIdStr = opeId + "_" + sdf.format(new Date());
//                                    jsonObjNew.put("dataSeqId", unqSeqIdStr);
//
//                                    jsonObjNew.put("defectFileUrl", "");
//                                    jsonObjNew.put("eqptId", "");
//                                    jsonObjNew.put("evtTimestamp", evtTimestampStr);
//                                    jsonObjNew.put("indexUrl", "");
//                                    jsonObjNew.put("opeId", opeId);
//                                    jsonObjNew.put("glassId", glassId);
//                                    jsonObjNew.put("ope_tbl_name", "system_alarm");
//                                    sendMessage("SYSTEM_ALARM", channel, jsonObjNew.toJSONString());
//                                    log.info("将找不到对应的配置文件的 QUEUE 放入SYSTEM_ALARM QUEUE 成功！ glassId={},opeId={} ", updateObj.getString("ope_tbl_name"), glassId, opeId);
////                                channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
////                                log.info("将本次 {} QUEUE 中需要等待的数据 ACK 成功！ glassId={},opeId={}", queueName, glassId, opeId);
//                                } catch (Exception e) {
//                                    log.error("ACK 原QUEUE 中需要等待的数据时失败！glassId={},opeId={} ", glassId, opeId);
//                                    log.error("ERROR message:", e);
//                                }
//                            } else {
//                                String opeId = updateObj.getString("ope_id");
//                                String glassId = updateObj.getString("glass_id");
//                                String evtTimestampStr = updateObj.getString("evt_timestamp");
//                                String jgeCnt = updateObj.getString("jge_cnt");
//                                try {
//                                    JSONObject jsonObjNew = new JSONObject();
//                                    jsonObjNew.put("dataInfo", String.format("opeId:%s glassId:%s evtTimestamp:%s jgeCnt:%s找不到对应的配置文件", opeId, glassId, evtTimestampStr, jgeCnt));
//                                    jsonObjNew.put("dataName", "");
//
//                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
//                                    String unqSeqIdStr = opeId + "_" + sdf.format(new Date());
//                                    jsonObjNew.put("dataSeqId", unqSeqIdStr);
//
//                                    jsonObjNew.put("defectFileUrl", "");
//                                    jsonObjNew.put("eqptId", "");
//                                    jsonObjNew.put("evtTimestamp", evtTimestampStr);
//                                    jsonObjNew.put("indexUrl", "");
//                                    jsonObjNew.put("opeId", opeId);
//                                    jsonObjNew.put("glassId", glassId);
//                                    jsonObjNew.put("ope_tbl_name", "system_alarm");
//                                    sendMessage("SYSTEM_ALARM", channel, jsonObjNew.toJSONString());
//                                    log.info("将找不到对应的配置文件的 QUEUE 放入SYSTEM_ALARM QUEUE 成功！ glassId={},opeId={} ", updateObj.getString("ope_tbl_name"), glassId, opeId);
////                                channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
////                                log.info("将本次 {} QUEUE 中需要等待的数据 ACK 成功！ glassId={},opeId={}", queueName, glassId, opeId);
//                                } catch (Exception e) {
//                                    log.error("ACK 原QUEUE 中需要等待的数据时失败！glassId={},opeId={} ", glassId, opeId);
//                                    log.error("ERROR message:", e);
//                                }
//                            }
//                        }
//
//                        //确定消费
//                        sureAck(channel, queueName, tableName, responseMap);
//                    } catch (Exception e) {
//                        LogUtil.pringStackTrace(e);
//                    }

                    //删除表锁
                    sureRedisAck(tableName);
                    return RETURN_FAISE;
                }
                HisConfigure hisConfigure = JacksonUtil.fromJson(strHisConfig, HisConfigure.class);
                if (null != hisConfigure) {
                    hisConfigure.setTableName(tableName);
                    if (hisConfigure.getPptFlg()) {
                        hisConfigure.setPptTableName(tableName);
                    }
                    if (hisConfigure.getHisFlg()) {
                        hisConfigure.setHisTableName(tableName);
                    }
                }

                //如果配置文件不是正确的格式，也直接送ERROR_QUEUE
                if (hisConfigure == null) {
                    log.error("table {}  config can not parse correctly", tableName);
//                    sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);

//                    try {
//                        for(JSONObject updateObj : jsonObjectList) {
//                            if (StringUtils.isNotEmpty(updateObj.getString("query_conditions"))) {
//                                JSONObject jsonObject = (JSONObject) JSONObject.parse(updateObj.getString("query_conditions"));
//                                String opeId = jsonObject.getString("ope_id");
//                                String glassId = jsonObject.getString("glass_id");
//                                String evtTimestampStr = jsonObject.getString("evt_timestamp");
//                                String jgeCnt = jsonObject.getString("jge_cnt");
//                                try {
//                                    JSONObject jsonObjNew = new JSONObject();
//                                    jsonObjNew.put("dataInfo", String.format("opeId:%s glassId:%s evtTimestamp:%s jgeCnt:%s的配置文件不是正确的格式", opeId, glassId, evtTimestampStr, jgeCnt));
//                                    jsonObjNew.put("dataName", "");
//
//                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
//                                    String unqSeqIdStr = opeId + "_" + sdf.format(new Date());
//                                    jsonObjNew.put("dataSeqId", unqSeqIdStr);
//
//                                    jsonObjNew.put("defectFileUrl", "");
//                                    jsonObjNew.put("eqptId", "");
//                                    jsonObjNew.put("evtTimestamp", evtTimestampStr);
//                                    jsonObjNew.put("indexUrl", "");
//                                    jsonObjNew.put("opeId", opeId);
//                                    jsonObjNew.put("glassId", glassId);
//                                    jsonObjNew.put("ope_tbl_name", "system_alarm");
//                                    sendMessage("SYSTEM_ALARM", channel, jsonObjNew.toJSONString());
//                                    log.info("将格式不正确的配置文件的 QUEUE 放入SYSTEM_ALARM QUEUE 成功！ glassId={},opeId={} ", updateObj.getString("ope_tbl_name"), glassId, opeId);
////                                channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
////                                log.info("将本次 {} QUEUE 中需要等待的数据 ACK 成功！ glassId={},opeId={}", queueName, glassId, opeId);
//                                } catch (Exception e) {
//                                    log.error("ACK 原QUEUE 中需要等待的数据时失败！glassId={},opeId={} ", glassId, opeId);
//                                    log.error("ERROR message:", e);
//                                }
//                            } else {
//                                String opeId = updateObj.getString("ope_id");
//                                String glassId = updateObj.getString("glass_id");
//                                String evtTimestampStr = updateObj.getString("evt_timestamp");
//                                String jgeCnt = updateObj.getString("jge_cnt");
//                                try {
//                                    JSONObject jsonObjNew = new JSONObject();
//                                    jsonObjNew.put("dataInfo", String.format("opeId:%s glassId:%s evtTimestamp:%s jgeCnt:%s的配置文件不是正确的格式", opeId, glassId, evtTimestampStr, jgeCnt));
//                                    jsonObjNew.put("dataName", "");
//
//                                    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
//                                    String unqSeqIdStr = opeId + "_" + sdf.format(new Date());
//                                    jsonObjNew.put("dataSeqId", unqSeqIdStr);
//
//                                    jsonObjNew.put("defectFileUrl", "");
//                                    jsonObjNew.put("eqptId", "");
//                                    jsonObjNew.put("evtTimestamp", evtTimestampStr);
//                                    jsonObjNew.put("indexUrl", "");
//                                    jsonObjNew.put("opeId", opeId);
//                                    jsonObjNew.put("glassId", glassId);
//                                    jsonObjNew.put("ope_tbl_name", "system_alarm");
//                                    sendMessage("SYSTEM_ALARM", channel, jsonObjNew.toJSONString());
//                                    log.info("将格式不正确的配置文件的 QUEUE 放入SYSTEM_ALARM QUEUE 成功！ glassId={},opeId={} ", updateObj.getString("ope_tbl_name"), glassId, opeId);
////                                channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
////                                log.info("将本次 {} QUEUE 中需要等待的数据 ACK 成功！ glassId={},opeId={}", queueName, glassId, opeId);
//                                } catch (Exception e) {
//                                    log.error("ACK 原QUEUE 中需要等待的数据时失败！glassId={},opeId={} ", glassId, opeId);
//                                    log.error("ERROR message:", e);
//                                }
//                            }
//                        }
//
//                        //确定消费
//                        sureAck(channel, queueName, tableName, responseMap);
//                    } catch (Exception e) {
//                        LogUtil.pringStackTrace(e);
//                    }

                    //删除表锁
                    sureRedisAck(tableName);
                    return RETURN_FAISE;
                }
                //endregion

                //region merge insert update
                String stageTableName = String.format("templates11_%s", tableName);
                String extTableName = String.format("ext_%s", tableName);

                if (!jsonObjectList.isEmpty()) {
                    if (hisConfigure.getPptFlg()) {
                        final String pptTableSchema = hisConfigure.getPptTableSchema();
                        final String pptTableName = hisConfigure.getPptTableName();
                        int effectRows = 0;
                        try {
                            effectRows = gploaderService.mergeProc(dbConnection, pptTableSchema, pptTableName, stageTableName, extTableName, jsonObjectList, limit_time, deleteInsert);
                        } catch (SQLException e) {
                            log.error("插入出现异常", e);

                            String exceptionErrorString = e.toString();

                            /**** GP 错误 ***************/
                            String connectionErrorString1 = "JDBCConnectionException";
                            String connectionErrorString2 = "Unable to acquire JDBC Connection";
                            String connectionErrorString3 = "This connection has been closed";
                            String connectionErrorString4 = "GenericJDBCException";
                            String connectionErrorString5 = "PersistenceException";
                            String connectionErrorString6 = "PoolExhaustedException";
                            String connectionErrorString7 = "connect timed out";
                            String connectionErrorString8 = "Unexpected internal error";
                            String connectionErrorString9 = "Out of memory";
                            String connectionErrorString13 = "CannotCreateTransactionException";
                            String connectionErrorString14 = "Unable to acquire JDBC Connection";
                            String connectionErrorString15 = "Could not get a resource from the pool";

                            /**** MQ 错误 ***************/
                            String connectionErrorString10 = "AmqpConnectException";
                            String connectionErrorString11 = "AmqpConnectException: java.net.ConnectException: Connection refused";
                            String connectionErrorString12 = "AmqpIOException";


                            /*** 增加GP/MQ连接报错的捕获,Queue不消费重新捞取 **/
                            if (exceptionErrorString.contains(connectionErrorString1) || exceptionErrorString.contains(connectionErrorString2)
                                    || exceptionErrorString.contains(connectionErrorString3) || exceptionErrorString.contains(connectionErrorString4)
                                    || exceptionErrorString.contains(connectionErrorString5) || exceptionErrorString.contains(connectionErrorString6)
                                    || exceptionErrorString.contains(connectionErrorString7) || exceptionErrorString.contains(connectionErrorString8)
                                    || exceptionErrorString.contains(connectionErrorString9) || exceptionErrorString.contains(connectionErrorString13)
                                    || exceptionErrorString.contains(connectionErrorString14) || exceptionErrorString.contains(connectionErrorString15)) {
                                log.error("GP连接异常，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            }else if (exceptionErrorString.contains(connectionErrorString10) || exceptionErrorString.contains(connectionErrorString11) || exceptionErrorString.contains(connectionErrorString12)) {
                                log.error("MQ连接异常，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            }//若表出现死锁情况，则将Queue Unack，释放锁，重新抢锁
                            else if (exceptionErrorString.indexOf("deadlock") > -1) {
                                log.error(pptTableName + "表出现死锁情况，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            } else {
                                sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
                                //确定消费
                                sureAck(channel, tableName, responseMap);
                            }
                        }
                        final String msg = "写入PPT表[" + pptTableName + "],记录数:" + effectRows;
                        stageAlreadyCreateFlg = true;
                        LogUtil.printPerformance(LOG_MERGE_PROC, queueName, jsonObjectList.size(), beginTime2);
                        beginTime2 = LogUtil.printCost(LOG_MERGE_PROC, queueName, msg, beginTime2);
                        stageTableName = String.format("templates11_%s", tableName);
                    }

                    log.info("==========================================");

                    if (hisConfigure.getHisFlg()) {
                        final String hisTableSchema = hisConfigure.getHisTableSchema();
                        final String hisTableName = hisConfigure.getHisTableName();
                        int effectRows = 0;
                        try {
                            effectRows = gploaderService.insertProc(dbConnection, hisTableSchema, hisTableName, stageTableName, extTableName, jsonObjectList, stageAlreadyCreateFlg);
                        } catch (SQLException e) {
                            log.error("插入出现异常", e);

                            String exceptionErrorString = e.toString();

                            /**** GP 错误 ***************/
                            String connectionErrorString1 = "JDBCConnectionException";
                            String connectionErrorString2 = "Unable to acquire JDBC Connection";
                            String connectionErrorString3 = "This connection has been closed";
                            String connectionErrorString4 = "GenericJDBCException";
                            String connectionErrorString5 = "PersistenceException";
                            String connectionErrorString6 = "PoolExhaustedException";
                            String connectionErrorString7 = "connect timed out";
                            String connectionErrorString8 = "Unexpected internal error";
                            String connectionErrorString9 = "Out of memory";
                            String connectionErrorString13 = "CannotCreateTransactionException";
                            String connectionErrorString14 = "Unable to acquire JDBC Connection";
                            String connectionErrorString15 = "Could not get a resource from the pool";

                            /**** MQ 错误 ***************/
                            String connectionErrorString10 = "AmqpConnectException";
                            String connectionErrorString11 = "AmqpConnectException: java.net.ConnectException: Connection refused";
                            String connectionErrorString12 = "AmqpIOException";


                            /*** 增加GP/MQ连接报错的捕获,Queue不消费重新捞取 **/
                            if (exceptionErrorString.contains(connectionErrorString1) || exceptionErrorString.contains(connectionErrorString2)
                                    || exceptionErrorString.contains(connectionErrorString3) || exceptionErrorString.contains(connectionErrorString4)
                                    || exceptionErrorString.contains(connectionErrorString5) || exceptionErrorString.contains(connectionErrorString6)
                                    || exceptionErrorString.contains(connectionErrorString7) || exceptionErrorString.contains(connectionErrorString8)
                                    || exceptionErrorString.contains(connectionErrorString9) || exceptionErrorString.contains(connectionErrorString13)
                                    || exceptionErrorString.contains(connectionErrorString14) || exceptionErrorString.contains(connectionErrorString15)) {
                                log.error("GP连接异常，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            }else if (exceptionErrorString.contains(connectionErrorString10) || exceptionErrorString.contains(connectionErrorString11) || exceptionErrorString.contains(connectionErrorString12)) {
                                log.error("MQ连接异常，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            }//若表出现死锁情况，则将Queue Unack，释放锁，重新抢锁
                            else if (exceptionErrorString.indexOf("deadlock") > -1) {
                                log.error(hisTableName + "表出现死锁情况，重新吃Queue", e);
                                //删除表锁
                                sureRedisAck(tableName);
                            } else {
                                sendJsonToErrorQueue(errorQueueRouteKey, jsonObjectList, updateList, realFlgList);
                                //确定消费
                                sureAck(channel, tableName, responseMap);
                            }
                        }
                        final String msg = "写入HIS表[" + hisTableName + "],记录数:" + effectRows;
                        LogUtil.printPerformance(LOG_SUB_MAIN_PROC_HIS, queueName, jsonObjectList.size(), beginTime2);
                        beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_HIS, queueName, msg, beginTime2);
                    }
                } else {
                    log.info("jsonObjectList is empty, updateList size is: {}.", updateList.size());
                }

                //update
                if (CollectionUtils.isNotEmpty(updateList)) {
                    final String pptTableSchema = hisConfigure.getPptTableSchema();
                    final String pptTableName = hisConfigure.getPptTableName();
                    final int effectRows = gploaderService.updateProc(dbConnection, jedis, updateList, pptTableSchema, pptTableName, stageTableName);
                    final String msg = "更新HIS表[" + pptTableName + "],记录数:" + effectRows;
                    LogUtil.printPerformance(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, updateList.size(), beginTime2);
                    beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, msg, beginTime2);
                }

                // 更新 real_flg 切换期间兼容
                if (CollectionUtils.isNotEmpty(realFlgList)) {
                    log.warn("进入了老的 real_flg 更新逻辑，请尽快修改，table: {}", tableName);
                    final int effectRows = gploaderService.updateProc(dbConnection, realFlgList);
                    final String msg = "更新HIS表[" + tableName + "],记录数:" + effectRows;
                    LogUtil.printPerformance(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, realFlgList.size(), beginTime2);
                    beginTime2 = LogUtil.printCost(LOG_SUB_MAIN_PROC_REAL_FLG, queueName, msg, beginTime2);
                }
                //endregion

                //确定消费
                // 这里其实可以拿到TableSchema， 但是如果同时有PPT/HIS两种配置的话程序需要调整，目前使用默认操作
                //PDS这部分，channel，queueName是正常的，tableName是细表名，不是pds_glass_summary
                sureAck(channel, queueName, tableName, responseMap);

                LogUtil.printPerformance(ACK, tableName, responseMap.size(), beginTime2);
                beginTime2 = LogUtil.printCost(ACK, queueName,
                        "sureAck:" + tableName + ", responseMap: " + responseMap.size(), beginTime2);

            } catch (SQLException sqlEx) {
                log.error("SQL exception: SQL State [{}], Error Code [{}], Error Message [{}].",
                        sqlEx.getSQLState(), sqlEx.getErrorCode(), sqlEx.getMessage());
                LogUtil.pringStackTrace(sqlEx);
                throw sqlEx;
            } catch (Exception ex3) {
                LogUtil.pringStackTrace(ex3);
                throw ex3;
            }

            log.info("table_name;{} total cost {} millis", tableName, System.currentTimeMillis() - tableProcStartTime);
            LogUtil.printPerformance(LOG_SUB_MAIN_PROC_SUM, tableName, allRowCount, tableProcStartTime);
            LogUtil.printCost(LOG_SUB_MAIN_PROC_SUM, tableName, " count:" + allRowCount, tableProcStartTime);

        } catch (SQLException sqlEx) {
            LogUtil.pringStackTrace(sqlEx);
            if (catchGpConnectionException(sqlEx.toString())) {
                log.error("[][][] 出现gp连接异常，Que不消费...");
            }
            //删除表锁
            sureRedisAck(tableName);
            throw sqlEx;
//            try {
//                if (catchGpConnectionException(sqlEx.toString())) {
//                    log.error("[][][] 出现gp连接异常，Que不消费...");
//                } else {
//                    sendJsonToErrorQueue(errorQueueRouteKey, queueName, tableName, sqlEx, null, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                }
//            } catch (Exception e) {
//                LogUtil.pringStackTrace(e);
//            }
        } catch (Exception ex) {
            /**
             * 如果一张表的数据没有处理完成，只送这一张表的ErrorQueue
             * 其余表对应的消息应该不受影响
             * 在此处送ERROR_QUEUE
             */
            LogUtil.pringStackTrace(ex);
            if (catchGpConnectionException(ex.toString())) {
                log.error("[][][] 出现gp连接异常，Que不消费...");
            }
            //删除表锁
            sureRedisAck(tableName);
            throw ex;
//            try {
//                if (catchGpConnectionException(ex.toString())) {
//                    log.error("[][][] 出现gp连接异常，Que不消费...");
//                } else {
//                    sendJsonToErrorQueue(errorQueueRouteKey, queueName, tableName, null, ex, jsonObjectList, updateList, realFlgList);
//                    //确定消费
//                    sureAck(channel, tableName, responseMap);
//                }
//            } catch (Exception e) {
//                LogUtil.pringStackTrace(e);
//            }
        }
        //endregion

        return RETURN_SUCCESS;
        //结束:打印耗时，写入行数等性能参数
    }


    /**
     * 无法处理的Q 扔进 HMS_ERROR2
     * 入口只有ERROR的Q，因为目前 Normal Q 处理异常时不知道具体是哪一笔报错不能都抛弃
     *
     * @param queueName
     * @param tableName
     * @param sqlEx
     * @return
     */
    private boolean needSendToError2(String queueName, String tableName, SQLException sqlEx) {
        // Error Q 抛弃
        if (StringUtils.isNotEmpty(queueName) && StringUtils.isNotEmpty(tableName)) {
            if (HMS_ERROR_Q.equals(queueName) || queueName.startsWith(HMS_ERROR_Q_BY_TABLE_PREFIX)) {
                /**
                 * ex instanceof org.postgresql.util.PSQLException
                 * tbl
                 *   teg
                 * error message
                 *  ERROR: numeric field overflow  (seg2 slice1 P1QMSTST01:40002 pid=18242)
                 *
                 * TODO
                 *  This rule should be setting in UI and save in redis, hms auto get all rule. hms not need to restart/update code when add new rule.
                 *
                 *
                 * String sqlState = sqlEx.getSQLState();
                 * int errorCode = sqlEx.getErrorCode();
                 * String errorMessage = sqlEx.getMessage();
                 */

//                if (null != sqlEx) {
//                    if (tegTables.contains(tableName) && SQL_STATE_NUMERIC_VALUE_OUT_OF_RANGE.equals(sqlEx.getSQLState())) {
//                        // ERROR: numeric field overflow  (seg2 slice1 P1QMSTST01:40002 pid=18242)
//                        return true;
//                    }
//                }
            }// end of from queue check: Just for from Error Q (All table)
        }

        return false;
    }

    private void sureRedisAck(String tableName) {
        try (Jedis jedis = jedisPool.getResource()) {
            String tableLockKey = String.format(REDIS_LOCK_KEY_FORMAT, tableName);
            jedis.del(tableLockKey);
            log.info("{} table: {}. 解锁[{}]成功.", REDIS_OPE, tableName, tableLockKey);
        } catch (Exception e) {
            log.error("Redis清除运行标志失败，原因{}", e.getMessage());
            LogUtil.pringStackTrace(e);
        }
    }

    /**
     * 异常送Error时的ack
     *
     * @param channel
     * @param tableName
     * @param responseMap
     */
    public void sureAck(Channel channel, String tableName, Map<GetResponse, String> responseMap) {
        sureAck(channel, null, tableName, responseMap);
        // queueName 可以拿到，这是非必要
    }

    public void sureAck(Channel channel, Map<GetResponse, String> responseMap) {
        ArrayList<GetResponse> responseList = new ArrayList<>();
        responseMap.forEach((ack, message) -> {
                responseList.add(ack);
        });
            responseList.forEach(ack -> {
                    try {
                        channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
                        //sendDefectQueInfoToRedis(jedis, ack, responseMap);
                    } catch (IOException e) {
                        LogUtil.pringStackTrace(e);
                    }
            });
        }


    /**
     * 正常存入DB后ack：queueName有值
     *
     * @param channel
     * @param queueName
     * @param tableName
     * @param responseMap
     */
    private void sureAck(Channel channel, String queueName, String tableName, Map<GetResponse, String> responseMap) {
        ArrayList<GetResponse> responseList = new ArrayList<>();

        //queueName为空时isErrorQ也为 false(即使是HMS_ERROR_*中的消息)
        // 因为此时是存数据异常送Error时的ack，不能清除redis中的异常记录
        // eg: Error Q 反复尝试不成功
        boolean isErrorQ = HmsUtil.isError(queueName);
        boolean isFileMessage = false;
        TableEntity tableEntity = isErrorQ ? getTableEntity(isFileMessage, tableName) : new TableEntity();
        List<String> primaryColumnList = isErrorQ ? TableJdbcUtil.getPrimaryColumnList(tableEntity) : new ArrayList<>();
        Map<String, String> colCameNameMap = isErrorQ ? getPrimaryColumnCamelMap(tableName, primaryColumnList) : new HashMap<>();
        List<String> errorMessageHashKeyList = new ArrayList<>();

        if (null == tableEntity) {
            log.error("sureAck 时获取 tableEntity {} 失败, queueName {}.", tableName, queueName);
        }

        responseMap.forEach((ack, message) -> {
            JSONObject jsonObject = new JSONObject();
            List<JSONObject> collect = new ArrayList<>();
            if (StringUtils.startsWith(message, "{")) {
                jsonObject = (JSONObject) JSONObject.parse(message);
            } else if (StringUtils.startsWith(message, "[")) {
                collect = JSONArray.parseArray(message).stream().map(obj -> (JSONObject) obj).collect(toList());
                if (collect != null && collect.size() > 0) {
                    jsonObject = collect.get(0);
                }
            }

            boolean pdsOrMdsIsTrue = "pds_glass_summary".equals(jsonObject.getString("ope_tbl_name")) || "mds_glass_summary".equals(jsonObject.getString("ope_tbl_name"));
            boolean summaryOrResultIsTrue = tableName.contains("summary") || tableName.contains("result");
            if (//只有 pds，mds glass 层和正常q会消费
                    (pdsOrMdsIsTrue || tableName.equals(jsonObject.getString("ope_tbl_name")))//pds,mds 符合条件
                            && (!summaryOrResultIsTrue))//不取pds,mds summary,result

//                if(tableName.equals(jsonObject.getString("ope_tbl_name")))
            {
                responseList.add(ack);
                if (isErrorQ) {
                    boolean isUpdateMsg = jsonObject.containsKey("update_flg") || jsonObject.containsKey("real_flg");
                    /**
                     * 数据Q
                     *  数组类型的Q（defect等）：只取第一笔即可 jsonObject
                     *  非数组类型的Q：循环 collect
                     * 更新Q
                     *  循环 collect
                     *
                     */

                    if (CollectionUtils.isNotEmpty(collect)) {
                        collect.forEach(j -> errorMessageHashKeyList.add(isFileMessage ? j.getString(UNQ_SEQ_ID) :
                                TableJdbcUtil.getColumnValString(j, primaryColumnList, colCameNameMap))
                        );
                    }

                }
            }
        });
//        final long lbeginTime = System.currentTimeMillis();


        if (!tableName.contains("summary")&&!tableName.contains("result")
                ) {//pds summary 不执行
            responseList.forEach(ack -> {
                try (Jedis jedis = jedisPool.getResource()) {
                    try {
                        channel.basicAck(ack.getEnvelope().getDeliveryTag(), false);
                        //sendDefectQueInfoToRedis(jedis, ack, responseMap);
                    } catch (IOException e) {
                            LogUtil.pringStackTrace(e);
                    }
                } catch (Exception e) {
                    if(!tableName.contains("mds")&&!tableName.contains("pds")) {
                        log.error(e.getMessage());
                        LogUtil.pringStackTrace(e);
                    }
                }
            });
        }

        //确认消费就确认删除此次运行标志
        sureRedisAck(tableName);

        //删除Error Q消息
        if (isErrorQ) {
            deleteErrorMessageDetail(queueName, tableName, errorMessageHashKeyList);
        }
    }





    /**
     * 送Error Q
     * 目前是配置文件不正确时从这里进入
     *
     * @param routeKey
     * @param jsonObjectList
     * @throws IOException
     * @throws TimeoutException
     */
    public void sendJsonToErrorQueue(String routeKey, List<JSONObject> jsonObjectList, List<JSONObject> updateList, List<JSONObject> realFlgList) throws IOException, TimeoutException {
        // 此时拿不到TableSchema，只有表名
        sendJsonToErrorQueue(routeKey, null, null, null, null, jsonObjectList, updateList, realFlgList);
    }

    /**
     * 发送Error Q
     *
     * @param routeKey
     * @param queueName
     * @param tableName
     * @param sqlEx
     * @param ex
     * @param jsonObjectList
     * @param updateList
     * @param realFlgList
     * @throws IOException
     * @throws TimeoutException
     */
    public void sendJsonToErrorQueue(String routeKey, String queueName, String tableName, SQLException sqlEx, Exception ex,
                                     List<JSONObject> jsonObjectList, List<JSONObject> updateList, List<JSONObject> realFlgList) throws IOException, TimeoutException {
        if (CollectionUtils.isEmpty(jsonObjectList) && CollectionUtils.isEmpty(updateList) && CollectionUtils.isEmpty(realFlgList)) {
            return;
        }

        if (StringUtils.isNotEmpty(tableName)) {
//            if (queueName.startsWith(A_DEFECT) || queueName.startsWith(HMS_ERROR_A_DEFECT_TABLE)) {
//                routeKey = createErrorQueue(A_DEFECT_TABLE);
//            }else if (queueName.startsWith(C_DEFECT) || queueName.startsWith(HMS_ERROR_C_DEFECT_TABLE)) {
//                routeKey = createErrorQueue(C_DEFECT_TABLE);
//            }else if (queueName.startsWith(F_DEFECT) || queueName.startsWith(HMS_ERROR_F_DEFECT_TABLE)) {
//                routeKey = createErrorQueue(F_DEFECT_TABLE);
//            }else {
            routeKey = createErrorQueue(tableName);
//            }
        }

        // data q
        if (CollectionUtils.isNotEmpty(jsonObjectList)) {
            List<String> errorJList = new ArrayList<>();
            if (1 == jsonObjectList.size() && needSendToError2(queueName, tableName, sqlEx)) {
                routeKey = errorQueueRouteKey2;
                errorJList.add(JSON.toJSONString(jsonObjectList));
            } else {
                errorJList = compressJsonObjectMessage(jsonObjectList);
            }
            sendStringToErrorQueue2(routeKey, errorJList);
            recordErrorMessageDetail(routeKey, tableName, sqlEx, ex, jsonObjectList);
            log.info("{} message compress to {}, from {} send to {}, table {}",
                    jsonObjectList.size(), errorJList.size(), queueName, routeKey, tableName);

        }

        // update q
        if (CollectionUtils.isNotEmpty(updateList) || CollectionUtils.isNotEmpty(realFlgList)) {
            List<JSONObject> errorUpdateJsonObjectList = new ArrayList<>();
            errorUpdateJsonObjectList.addAll(updateList);
            errorUpdateJsonObjectList.addAll(realFlgList);

            // Add for save redis record
            errorUpdateJsonObjectList.forEach(j -> {
                if (null == j.getString(UNQ_SEQ_ID)) {
                    j.put(UNQ_SEQ_ID, UUID.randomUUID().toString());
                }
            });

            List<String> errorJList = compressJsonObjectMessage(errorUpdateJsonObjectList);
            recordErrorMessageDetail(routeKey, tableName, sqlEx, ex, errorUpdateJsonObjectList, true);
            sendStringToErrorQueue2(routeKey, errorJList);
            log.info("{} update message compress to {}, from {} send to {}, table {}",
                    errorUpdateJsonObjectList.size(), errorJList.size(), queueName, routeKey, tableName);
        }
    }

    /**
     * 获取主键的列名vs驼峰Map
     *
     * @param tableName
     * @param primaryColumnList
     * @return
     */
    private static Map<String, String> getPrimaryColumnCamelMap(String tableName, List<String> primaryColumnList) {
        Map<String, String> colCameNameMap = primaryColumnCamelMap.get(tableName);
        if (null == colCameNameMap) {
            Map<String, String> colCameNameMap2 = new HashMap<>();
            primaryColumnList.forEach(c -> colCameNameMap2.put(c, TableJdbcUtil.underline2Camel(c)));
            primaryColumnCamelMap.put(tableName, colCameNameMap2);
            return colCameNameMap2;
        }
        return colCameNameMap;
    }

    /**
     * 获取表结构，这些结构在报错前已经缓存
     * 存储error信息时使用
     * 这里取table schema不方便，先尝试sor，再尝试dm
     *
     * @param isFileMessage
     * @param tableName
     * @return
     */
    private static TableEntity getTableEntity(boolean isFileMessage, final String tableName) {
        if (isFileMessage) {
            return TableMetaService.getTableMetaInfo(TABLE_SCHEMA_DEFECT, tableName);
        } else {
            TableEntity tableEntity = TableMetaService.getTableMetaInfo(TABLE_SCHEMA_SOR, tableName);
            if (null == tableEntity) {
                tableEntity = TableMetaService.getTableMetaInfo(TABLE_SCHEMA_DM, tableName);
                if (null == tableEntity) {
                    tableEntity = TableMetaService.getTableMetaInfo(TABLE_SCHEMA_EDA, tableName);
                    if (null == tableEntity) {
                        log.error("记录error信息时获取 {} 表结构失败.", tableName);
                        return null;
                    }
                }
            }

            return tableEntity;
        }
    }

    /**
     * 的喊出redis中的error信息
     *
     * @param queueName
     * @param tableName
     * @param errorMessageHashKeyList
     */
    private void deleteErrorMessageDetail(final String queueName, final String tableName, final List<String> errorMessageHashKeyList) {
        final String tableKey = String.format(REDIS_ERROR_MESSAGE_KEY_FORMAT, queueName);
        try (Jedis jedis = jedisPool.getResource()) {
            log.info("{} delete {} error from redis [{}]-{}, queueName [{}], tableName [{}].",
                    REDIS_OPE, (errorMessageHashKeyList.isEmpty() ? 0L : jedis.hdel(tableKey, errorMessageHashKeyList.toArray(new String[errorMessageHashKeyList.size()]))), tableKey, errorMessageHashKeyList, queueName, tableName);
        } catch (Exception e) {
            log.error("{} delete error from redis failed: {}! errorMessageHashKeyList-{}", REDIS_OPE, e.getMessage(), errorMessageHashKeyList);
            LogUtil.pringStackTrace(e);
        }
    }

    /**
     * 存储error明细到redis
     * 数组类消息使用
     *
     * @param routeKey
     * @param tableName
     * @param sqlEx
     * @param ex
     * @param jsonObjMap
     */
//    private void recordErrorMessageDetail(final String routeKey, final String tableName,
//                                          final SQLException sqlEx, final Exception ex,
//                                          final Map<String, List<JSONObject>> jsonObjMap) {
//        jsonObjMap.keySet().forEach(key -> {
//            List<JSONObject> jsonObjectList = jsonObjMap.get(key);
//            if (CollectionUtils.isNotEmpty(jsonObjectList)) {
//                List<JSONObject> oneJsonObjectList = Collections.singletonList(jsonObjectList.get(0));
//                recordErrorMessageDetail(routeKey, tableName, sqlEx, ex, oneJsonObjectList);
//            }
//        });
//    }

    /**
     * 存储error明细到redis
     * 非update sql
     *
     * @param routeKey
     * @param tableName
     * @param sqlEx
     * @param ex
     * @param jsonObjList
     */
    private void recordErrorMessageDetail(final String routeKey, final String tableName,
                                          final SQLException sqlEx, final Exception ex,
                                          final List<JSONObject> jsonObjList) {
        recordErrorMessageDetail(routeKey, tableName, sqlEx, ex, jsonObjList, false);
    }

    /**
     * 存储error明细到redis
     *
     * @param routeKey
     * @param tableName
     * @param sqlEx
     * @param ex
     * @param jsonObjList
     * @param isUpdateMessage
     */
    private void recordErrorMessageDetail(final String routeKey, final String tableName,
                                          final SQLException sqlEx, final Exception ex,
                                          final List<JSONObject> jsonObjList, boolean isUpdateMessage) {
        if (StringUtils.isEmpty(tableName)) {
            return;
        }
        boolean isFileMessage = false;
        final TableEntity tableEntity = getTableEntity(isFileMessage, tableName);
        if (null == tableEntity) {
            return;
        }

        final String tableKey = String.format(REDIS_ERROR_MESSAGE_KEY_FORMAT, routeKey);
        final String errorCode = null != sqlEx ? sqlEx.getSQLState() : "";
        final String errorMessage = null != sqlEx ? sqlEx.getMessage() : (ex != null ? ex.getMessage() : "");
        final Timestamp curTime = Timestamp.valueOf(LocalDateTime.now());

        final List<String> primaryColumnList = TableJdbcUtil.getPrimaryColumnList(tableEntity);
        final Map<String, String> colCameNameMap = getPrimaryColumnCamelMap(tableName, primaryColumnList);

        try (Jedis jedis = jedisPool.getResource()) {
            if (isUpdateMessage) {
                jsonObjList.forEach(j -> {
                    ErrorMessageInfoEntity errorInfo = new ErrorMessageInfoEntity();
                    errorInfo.setErrorCode(errorCode);
                    errorInfo.setErrorMessage(errorMessage);
                    errorInfo.setUpdateSql(JSON.toJSONString(j));
                    errorInfo.setSendToErrorQueueTimestamp(curTime);

                    jedis.hset(tableKey, j.getString(UNQ_SEQ_ID), JSON.toJSONString(errorInfo));
                });
            } else {
                jsonObjList.forEach(j -> {
                    ErrorMessageInfoEntity errorInfo = new ErrorMessageInfoEntity();
                    errorInfo.setErrorMessage(errorMessage);
                    errorInfo.setSendToErrorQueueTimestamp(curTime);
                    errorInfo.setPkInfo(TableJdbcUtil.getColumnMap(j, primaryColumnList, colCameNameMap));

                    jedis.hset(tableKey, TableJdbcUtil.getColumnValString(j, primaryColumnList, colCameNameMap), JSON.toJSONString(errorInfo));
                });
            }
            log.info("{} save {} error info to redis [{}], routeKey [{}], tableName [{}] ",
                    REDIS_OPE, jsonObjList.size(), tableKey, routeKey, tableName);
        } catch (Exception e) {
            log.error("{} save error info to redis failed: {}!", REDIS_OPE, e.getMessage());
            LogUtil.pringStackTrace(e);
        }
    }


    private List<String> compressJsonObjectMessage(List<JSONObject> jsonList) {
        ArrayList<String> errorJList = new ArrayList<>();
        int jsonSize = jsonList.size();
        if (jsonSize == 1) {
            errorJList.add(JSONArray.toJSONString(jsonList));
        } else {
            //分组array
            if (jsonSize > 10000) {
                List<JSONObject> fourList = jsonList.subList(0, jsonSize / 4);
                List<JSONObject> thirdList = jsonList.subList(jsonSize / 4, (jsonSize / 4) * 2);
                List<JSONObject> halfList = jsonList.subList((jsonSize / 4) * 2, (jsonSize / 4) * 3);
                List<JSONObject> oneList = jsonList.subList((jsonSize / 4) * 3, jsonSize);
                errorJList.add(JSONArray.toJSONString(fourList));
                errorJList.add(JSONArray.toJSONString(halfList));
                errorJList.add(JSONArray.toJSONString(thirdList));
                errorJList.add(JSONArray.toJSONString(oneList));
            } else if (jsonSize >= 3000) {
                List<JSONObject> thirdList = jsonList.subList(0, jsonSize / 3);
                List<JSONObject> halfList = jsonList.subList(jsonSize / 3, (jsonSize / 3) * 2);
                List<JSONObject> oneList = jsonList.subList((jsonSize / 3) * 2, jsonSize);
                errorJList.add(JSONArray.toJSONString(thirdList));
                errorJList.add(JSONArray.toJSONString(halfList));
                errorJList.add(JSONArray.toJSONString(oneList));
            } else {
                List<JSONObject> halfList = jsonList.subList(0, jsonSize / 2);
                List<JSONObject> oneList = jsonList.subList(jsonSize / 2, jsonSize);
                errorJList.add(JSONArray.toJSONString(halfList));
                errorJList.add(JSONArray.toJSONString(oneList));
            }
        }
        return errorJList;
    }

//    private void sendStringToErrorQueue2(String routeKey, String jsonString) throws IOException, TimeoutException {
//        if (StringUtils.isEmpty(jsonString)) {
//            return;
//        }
//        try (Connection connection = connectionFactory.newConnection()) {
//            final Channel channel = connection.createChannel();
//            sendMessage(routeKey, channel, jsonString);
//        }
//    }

    private void sendStringToErrorQueue2(String routeKey, List<String> list) throws IOException, TimeoutException {
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        try (Connection connection = connectionFactory.newConnection()) {
            final Channel channel = connection.createChannel();
            list.forEach(message -> sendMessage(routeKey, channel, message));
        } catch (Exception e) {
            LogUtil.pringStackTrace(e);
            throw new HmsException(e);
        }
    }

    private void sendMessage(String routeKey, Channel channel, String message) {
        try {
            channel.basicPublish(exchange, routeKey,
                    new AMQP.BasicProperties.Builder()
                            .contentType("text/plain")
                            .deliveryMode(2)//设置消息持久性:2永不过期
                            .priority(1).build(), message.getBytes());
        } catch (IOException e) {
            LogUtil.pringStackTrace(e);
            throw new HmsException(e);
        }
    }

    /**
     * 创建Q & 绑定队列
     * 目前只针对Error Q
     *
     * @param tableName
     */
    private String createErrorQueue(String tableName) {
        try (Connection connection = connectionFactory.newConnection()) {
            final Channel channel = connection.createChannel();

            if (tableName.startsWith(EDC_DATA_TABLE_PREFIX)) {
                tableName = EDC_DATA_TABLE_ERROR_Q_SUFFIX;
            }

            String queueName = HMS_ERROR_Q_BY_TABLE_PREFIX + tableName;
            String routeKey = queueName;
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, exchange, routeKey);
            return queueName;
        } catch (Exception e) {
            LogUtil.pringStackTrace(e);
            throw new HmsException(e);
        }
    }

    /**
     * Check 是否是Gp连接错误
     *
     * @param exceptionErrorString
     * @return
     */
    private boolean catchGpConnectionException(String exceptionErrorString) {
        String connectionErrorString1 = "JDBCConnectionException";
        String connectionErrorString2 = "Unable to acquire JDBC Connection";
        String connectionErrorString3 = "This connection has been closed";
        String connectionErrorString4 = "GenericJDBCException";
        String connectionErrorString5 = "PersistenceException";
        String connectionErrorString6 = "PoolExhaustedException";
        String connectionErrorString7 = "connect timed out";
        String connectionErrorString8 = "Unexpected internal error";
        String connectionErrorString9 = "Out of memory";
        if (exceptionErrorString.contains(connectionErrorString1) || exceptionErrorString.contains(connectionErrorString2)
                || exceptionErrorString.contains(connectionErrorString3) || exceptionErrorString.contains(connectionErrorString4)
                || exceptionErrorString.contains(connectionErrorString5) || exceptionErrorString.contains(connectionErrorString6)
                || exceptionErrorString.contains(connectionErrorString7) || exceptionErrorString.contains(connectionErrorString8)
                || exceptionErrorString.contains(connectionErrorString9)) {
            return true;
        }
        return false;
    }

}